package com.dy.realtime.dwd;

import com.dy.realtime.common.base.BaseSqlApp;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.SQLUtil;
import com.sun.org.apache.bcel.internal.Const;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.realtime.dwd
 * @className: DwdTradeConsultation
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/10 19:41
 * @version: 1.0
 *  todo 交易域问诊事务事实表
 */

public class DwdTradeConsultation extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeConsultation().start(Constant.TOPIC_DB_Medical,Constant.TOPIC_DB_Medical,4,10021);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        readOdsDb(tableEnv,Constant.TOPIC_DWD_TRAFFIC_CONSULTATION);
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] `id`,\n" +
                "`data`['create_time'] `consultation_time`,\n" +
                "`data`['consultation_fee'] `consultation_fee`,\n" +
                "`data`['doctor_id'] `doctor_id`,\n" +
                "`data`['patient_id'] `patient_id`,\n" +
                "`data`['user_id'] `user_id`\n" +
                "from topic_db_medical   \n" +
                "where `table` = 'consultation'\n" +
                "and `type` = 'insert'");

//        resultTable.execute().print();

        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRAFFIC_CONSULTATION+" (\n" +
                "    id string,\n" +
                "    consultation_time string,\n" +
                "    consultation_fee string,\n" +
                "    doctor_id string,\n" +
                "    patient_id string,\n" +
                "    user_id string\n" +
                ")"+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRAFFIC_CONSULTATION));
        resultTable.executeInsert(Constant.TOPIC_DWD_TRAFFIC_CONSULTATION);


    }
}
